/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.prediction.top;

import aiaudio.processing.base.mapreduce.CoordinateDoubleValueKey;
import aiaudio.database.tables.RecommendationListTable;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author nastya
 */
public class MakeRecommendationReducer extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable> {
    public static final int DEFAULT_MAX_POSITION = 30;
    public static final String MAX_POSITION_PARAM = "maxPosition";
    
    private int maxPosition;

    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<CoordinateDoubleValueKey> values, Context context) throws IOException, InterruptedException {
        TopRatingList l = new TopRatingList(maxPosition);
        for (CoordinateDoubleValueKey coordinateDoubleValueKey : values) {
            l.add(coordinateDoubleValueKey);
        }
        
        Put put = new Put(key.get());
        System.out.println(Bytes.toString(key.get()));
        fill(l, put);
        context.write(new ImmutableBytesWritable(key), put);
    }

    protected void fill(TopRatingList l, Put put) {
        for (int i = 0; i < l.size(); i++) {
            CoordinateDoubleValueKey get = l.get(i);
            System.out.println(i + ") " + get.getValue());
            RecommendationListTable.addRecommendation(put, i, get.getCoordinate(), get.getValue());
        }
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        maxPosition = context.getConfiguration().getInt(MAX_POSITION_PARAM, DEFAULT_MAX_POSITION);
    }
    
}
